package org.comanche.snaf.core.polling;

import java.net.SocketAddress;
import java.util.Iterator;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicReference;

import org.comanche.snaf.core.handler.SessionHandler;
import org.comanche.snaf.core.service.Acceptor;
import org.comanche.snaf.core.service.Processor;
import org.comanche.snaf.core.session.IoSession;

/**
 * This is an abstract class using the polling strategy
 * to accept the connect request and create new session to 
 * polling process. Currently it only create one address to bind
 * 
 * @author joe_zhou
 *
 */
public abstract class AbstractPollingAcceptor<S extends IoSession, H> implements Acceptor{

	//////////////////////////////////////////////////////////
	// property members
	// we use JDLK's thread pool to manage thread
	private Executor executor;
	// the local address which will be listened
	private SocketAddress localAdderss;
	// the flag is use to decide the address can be reused
	private boolean isReuseAddress = false;
	// the handler which control the session
	private SessionHandler handler;
	// the atomic reference
	private AtomicReference<Acceptor> acceptorRef = new AtomicReference<Acceptor>();
	// the processor of session
	private Processor<S> processor;
	// the backlog
	private int backlog = 50;
	
	//////////////////////////////////////////////////////////
	// constructor
	public AbstractPollingAcceptor(Processor<S> processor){
		this.processor = processor;
	}
	
	///////////////////////////////////////////////////////
	// protected method
	/**
	 * bind the local address by extends class
	 */
	protected abstract void bindInternal(SocketAddress localAddress) throws Exception;
	/**
	 * get acceptable connect
	 * @return
	 * @throws Exception
	 */
	protected abstract int select() throws Exception;
	/**
	 * 
	 */
	protected abstract void wakeup();
	/**
	 * get the accept requests from the client send
	 * @param hanlders
	 * @return
	 */
	protected abstract S accept(Processor<S> processor, H handler) throws Exception;
	/**
	 * 
	 * @return
	 */
	protected abstract Iterator<H> getHandlers();
	
	protected int getBacklog(){
		return backlog;
	}
	
	///////////////////////////////////////////////////////
	// inner class
	/**
	 * it will listen all the accept request and 
	 * make it in request queue when new request come in, it will
	 * create new session and related it with processor
	 */
	private class Acceptor implements Runnable {
		
		public void run() {
			while (true) {
				try {
					int select = select();
					// there is some connect request
					if (select > 0){
						process(getHandlers());
					}
					
					
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
		
	}
	
	private void process(Iterator<H> handlers){
		// iterator the handlers and create session
		// with the processor
		while (handlers.hasNext()){
			try {
				H handler = handlers.next();
				handlers.remove();
				// create session
				S session = accept(processor, handler);
				if (session != null) {
					// add new session
					processor.add(session);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
	}
	
	///////////////////////////////////////////////////////
	// interface
	public boolean isReuseAddress() {
		return isReuseAddress;
	}
	
	public void setReuseAddress(boolean reuseAddress) {
		isReuseAddress = reuseAddress;	
	}
	
	public void bind() {
		bind(getDefaultLocalAddress());	
	}
	
	public void bind(SocketAddress localAddress) {
		if (localAddress == null) {
			throw new IllegalArgumentException("localAddress is null");
		}
		
		if (acceptorRef.get() == null){
			// bind the local address first
			try {
				bindInternal(localAddress);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				throw new RuntimeException(e);
			}
			// if the acceptor job has been ran, then just skip it
			Acceptor acceptor = new Acceptor();
			// run the job if it set success
			if (acceptorRef.compareAndSet(null, acceptor)){
				executor.execute(acceptor);
			}
			// wake up the select thread
			wakeup();
		}
	}
	
	/**
	 * we will not implements the unbind function
	 */
	public void unbind() {
		throw new RuntimeException("Not implement issue");
	}
	/**
	 * we will not implements the unbind function
	 */
	public void unbind(SocketAddress localAddress) {
		throw new RuntimeException("Not implement issue");
	}
	/**
	 * 
	 */
	public void setDefaultLocalAddress(SocketAddress localAddress) {
		this.localAdderss = localAddress;
		
	}
	/**
	 * 
	 */
	public void setHandler(SessionHandler handler) {
		this.handler = handler;
	}
	/**
	 * 
	 */
	public SessionHandler getHandler() {
		return handler;
	}
	/**
	 * 
	 */
	public SocketAddress getDefaultLocalAddress() {
		return localAdderss;
	}
}
